Please address the comments from this code review:

## Overall Comments
- The minimum-timeout enforcement logic (`if timeout < 30: timeout = 30`) is duplicated across many tools (`fetch_url`, Tavily/BoCha/Exa helpers, etc.); consider extracting a small utility (or a module-level `MIN_TIMEOUT` constant plus helper) to centralize this behavior and avoid inconsistencies (e.g., `_web_search_exa` currently lacks the clamp).
- The Exa API key missing error message is in Chinese in `_get_exa_key` while other user-facing errors in this module are English; aligning these messages to a consistent language will make debugging and UX more coherent.
- The lists of supported web-search tools for reference extraction are now duplicated in multiple places (e.g., `astr_agent_hooks._extract_web_search_refs`, dashboard routes, and `MessageList.vue`), which makes it easy to miss a spot when adding new providers; consider centralizing this mapping or deriving it from a shared config to keep UI and backend behavior in sync.

## Individual Comments

### Comment 1
<location path="astrbot/builtin_stars/web_searcher/main.py" line_range="159-168" />
<code_context>
         self,
         cfg: AstrBotConfig,
         payload: dict,
+        timeout: int = 30,
     ) -> list[SearchResult]:
         """使用 Tavily 搜索引擎进行搜索"""
</code_context>
<issue_to_address>
**suggestion:** Normalize the timeout value inside `_web_search_exa` for consistency and safety.

Other helpers (`_get_from_url`, `_web_search_tavily`, `_extract_tavily`, `_web_search_bocha`, `_extract_exa`, `_find_similar_exa`) all enforce a minimum 30s timeout internally, while `_web_search_exa` relies on its caller (`search_from_exa`) to clamp the value. If `_web_search_exa` is reused elsewhere, it may see much smaller timeouts and behave inconsistently. Please add the same `if timeout < 30: timeout = 30` guard at the top of `_web_search_exa` to align behavior and avoid unexpectedly short timeouts.

Suggested implementation:

```python
    ) -> list[SearchResult]:
        """使用 Exa 搜索引擎进行搜索"""
        if timeout < 30:
            timeout = 30

```

If the `_web_search_exa` signature or docstring differ slightly (e.g., different Chinese text or no docstring), adjust the SEARCH pattern to match the actual function header and insert:

    if timeout < 30:
        timeout = 30

as the first statement in the function body, immediately after any docstring, to keep behavior consistent with `_get_from_url`, `_web_search_tavily`, `_extract_tavily`, `_web_search_bocha`, `_extract_exa`, and `_find_similar_exa`.
</issue_to_address>

### Comment 2
<location path="astrbot/builtin_stars/web_searcher/main.py" line_range="68" />
<code_context>
         """清理文本，去除空格、换行符等"""
         return text.strip().replace("\n", " ").replace("\r", " ").replace("  ", " ")

-    async def _get_from_url(self, url: str) -> str:
+    async def _get_from_url(self, url: str, timeout: int = 30) -> str:
         """获取网页内容"""
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared helpers for timeout handling, base-URL construction, and Exa HTTP requests to remove duplication and make the web search code easier to maintain.

You can keep all the new functionality while cutting a lot of duplication with a few small helpers. The main hot spots are timeout handling, base URL construction, and Exa HTTP calls.

### 1. Centralize timeout normalization

The `if timeout < 30: timeout = 30` pattern is repeated many times.

Add a helper:

```python
def _normalize_timeout(self, timeout: int | None, minimum: int = 30) -> aiohttp.ClientTimeout:
    if timeout is None:
        timeout = minimum
    elif timeout < minimum:
        timeout = minimum
    return aiohttp.ClientTimeout(total=timeout)
```

Then use it at call sites instead of repeating the logic:

```python
async def _web_search_tavily(self, cfg: AstrBotConfig, payload: dict, timeout: int = 30) -> list[SearchResult]:
    tavily_key = await self._get_tavily_key(cfg)
    base_url = self._tavily_base_url(cfg)
    url = f"{base_url}/search"
    header = {
        "Authorization": f"Bearer {tavily_key}",
        "Content-Type": "application/json",
    }
    timeout_obj = self._normalize_timeout(timeout)
    async with aiohttp.ClientSession(trust_env=True) as session:
        async with session.post(url, json=payload, headers=header, timeout=timeout_obj) as response:
            ...
```

And for tools you can drop the inline clamp:

```python
@llm_tool(name="fetch_url")
async def fetch_website_content(self, event: AstrMessageEvent, url: str, timeout: int = 30) -> str:
    timeout_obj = self._normalize_timeout(timeout)
    resp = await self._get_from_url(url, timeout_obj.total)
    return resp
```

(or just pass `timeout_obj` through if you adjust `_get_from_url`).

### 2. Extract base URL helpers for providers

The Tavily and Exa base URL logic is repeated.

Add:

```python
def _tavily_base_url(self, cfg: AstrBotConfig) -> str:
    return (
        cfg.get("provider_settings", {})
        .get("websearch_tavily_base_url", "https://api.tavily.com")
        .rstrip("/")
    )

def _exa_base_url(self, cfg: AstrBotConfig) -> str:
    return (
        cfg.get("provider_settings", {})
        .get("websearch_exa_base_url", "https://api.exa.ai")
        .rstrip("/")
    )
```

Then simplify call sites:

```python
base_url = self._tavily_base_url(cfg)
url = f"{base_url}/search"
```

```python
base_url = self._exa_base_url(cfg)
url = f"{base_url}/contents"
```

This removes duplication and keeps provider-specific config in one place.

### 3. Consolidate Exa HTTP request logic

`_web_search_exa`, `_extract_exa`, and `_find_similar_exa` all repeat the same HTTP boilerplate. You can pull that out into one internal helper that deals with key retrieval, base URL, headers, timeout, and error handling:

```python
async def _exa_request(
    self,
    cfg: AstrBotConfig,
    path: str,
    payload: dict,
    timeout: int = 30,
) -> dict:
    exa_key = await self._get_exa_key(cfg)
    base_url = self._exa_base_url(cfg)
    url = f"{base_url}/{path.lstrip('/')}"
    header = {
        "x-api-key": exa_key,
        "Content-Type": "application/json",
    }
    timeout_obj = self._normalize_timeout(timeout)

    async with aiohttp.ClientSession(trust_env=True) as session:
        async with session.post(url, json=payload, headers=header, timeout=timeout_obj) as response:
            if response.status != 200:
                reason = await response.text()
                raise Exception(
                    f"Exa request to {path} failed: {reason}, status: {response.status}",
                )
            return await response.json()
```

Then each high-level method only shapes payload and maps results:

```python
async def _web_search_exa(
    self,
    cfg: AstrBotConfig,
    payload: dict,
    timeout: int = 30,
) -> list[SearchResult]:
    data = await self._exa_request(cfg, "search", payload, timeout=timeout)
    results: list[SearchResult] = []
    for item in data.get("results", []):
        results.append(
            SearchResult(
                title=item.get("title", ""),
                url=item.get("url", ""),
                snippet=(item.get("text") or "")[:500],
            )
        )
    return results
```

```python
async def _extract_exa(
    self, cfg: AstrBotConfig, payload: dict, timeout: int = 30
) -> list[dict]:
    data = await self._exa_request(cfg, "contents", payload, timeout=timeout)
    results: list[dict] = data.get("results", [])
    if not results:
        raise ValueError("Error: Exa content extraction does not return any results.")
    return results
```

```python
async def _find_similar_exa(
    self, cfg: AstrBotConfig, payload: dict, timeout: int = 30
) -> list[SearchResult]:
    data = await self._exa_request(cfg, "findSimilar", payload, timeout=timeout)
    results: list[SearchResult] = []
    for item in data.get("results", []):
        results.append(
            SearchResult(
                title=item.get("title", ""),
                url=item.get("url", ""),
                snippet=(item.get("text") or "")[:500],
            )
        )
    return results
```

This way, if you change headers, auth, or error handling, you only touch `_exa_request`.

### 4. Optional: small helpers for repeated validation

If you want to further simplify the public tool methods, a couple of tiny validators can keep them focused on behavior rather than plumbing.

For example, Exa config check and clamping:

```python
def _ensure_exa_config(self, cfg: AstrBotConfig) -> None:
    if not cfg.get("provider_settings", {}).get("websearch_exa_key", []):
        raise ValueError("Error: Exa API key is not configured in AstrBot.")

def _clamp_results(self, value: int, minimum: int, maximum: int) -> int:
    return max(minimum, min(value, maximum))
```

Usage:

```python
@llm_tool("web_search_exa")
async def search_from_exa(..., max_results: int = 10, ...):
    ...
    cfg = self.context.get_config(umo=event.unified_msg_origin)
    self._ensure_exa_config(cfg)

    max_results = self._clamp_results(max_results, 1, 100)
    ...
```

These changes keep all functionality (timeouts, base URLs, Exa/Tavily features) but reduce repetition and make future changes safer and easier.
</issue_to_address>

Please address the comments from this code review:

## Overall Comments
- WEB_SEARCH_REFERENCE_TOOLS is now defined separately in Python (web_search_utils.py) and in the frontend (MessageList.vue); consider centralizing this list or generating the frontend list from the backend to avoid drift when adding/removing tools.
- _normalize_timeout unconditionally casts timeout to int, so values like 30.9 or large numeric strings will be truncated rather than validated; if you expect non-integer or very large values, you might want stricter range checking and clearer error handling instead of silently coercing.

## Individual Comments

### Comment 1
<location path="astrbot/core/tools/web_search_tools.py" line_range="648-657" />
<code_context>
+        max_results = max(1, min(int(kwargs.get("max_results", 10)), 100))
</code_context>
<issue_to_address>
**issue (bug_risk):** Unvalidated int() casts for max_results can raise on malformed input and abort tool execution.

Both `ExaWebSearchTool` and `ExaFindSimilarTool` call `int(kwargs.get("max_results", 10))` directly, so a non-integer value (e.g. "10.5", "foo", or an unexpected type) will raise `ValueError` and abort the tool call. To match how other user-supplied numerics are handled, consider wrapping this in a helper or local try/except that falls back to 10 and then applies the 1–100 clamp, so invalid input degrades gracefully instead of throwing.
</issue_to_address>

### Comment 2
<location path="tests/unit/test_web_search_utils.py" line_range="96-105" />
<code_context>
+    assert [ref["index"] for ref in refs["used"]] == ["a152.1", "a152.2"]
+
+
+@pytest.mark.parametrize(
+    ("base_url", "expected_message"),
+    [
+        (
+            "exa.ai/search",
+            "Error: Exa API Base URL must start with http:// or https://. "
+            "Proxy base paths are allowed. Received: 'exa.ai/search'.",
+        ),
+    ],
+)
+def test_normalize_web_search_base_url_reports_invalid_value(
+    base_url: str, expected_message: str
+):
+    with pytest.raises(ValueError) as exc_info:
+        normalize_web_search_base_url(
+            base_url,
+            default="https://api.exa.ai",
+            provider_name="Exa",
+        )
+
</code_context>
<issue_to_address>
**suggestion (testing):** Extend normalize_web_search_base_url tests to cover empty/None values and multiple invalid schemes.

Key branches still aren’t covered:

- `base_url` is `None` or an empty string → should return the trimmed `default` value. Please add a parametrized test for this.
- Additional invalid values like `"ftp://exa.ai"`, `"https:///only-path"`, or `"http://"` (no netloc) → should raise `ValueError` with the same message format.

These cases will complete coverage for scheme/netloc validation and default handling.
</issue_to_address>

Please address the comments from this code review:

## Overall Comments
- The allowed Exa `search_type` values and `category` options are hard-coded in both `_EXA_SEARCH_TYPES` and the `ExaWebSearchTool.parameters` description; consider centralizing these in shared constants (and reusing in the schema description) to avoid drift between validation and documentation.
- `normalize_web_search_base_url` can now raise `ValueError` during request construction (e.g., `_get_exa_base_url`, `_get_tavily_base_url`, `URLExtractor.__init__`); if you expect misconfiguration in production, consider catching this closer to configuration load or surfacing a clearer, provider-level error instead of letting it bubble up as a runtime exception.

## Individual Comments

### Comment 1
<location path="tests/unit/test_web_search_utils.py" line_range="96-97" />
<code_context>
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("search_type", "expected"),
+    [
</code_context>
<issue_to_address>
**suggestion (testing):** Add coverage for defaulting behavior of `normalize_web_search_base_url` when `base_url` is `None` or empty.

Current tests cover invalid schemes and disallowed suffixes. Please also add a parametrized test that passes `base_url=None` and `base_url="  "` and asserts that `normalize_web_search_base_url` returns the normalized `default` value. This will ensure callers like `_get_tavily_base_url`, `_get_exa_base_url`, and `URLExtractor` can rely on the default when the field is left blank.

```suggestion
@pytest.mark.parametrize(
    ("base_url", "default"),
    [
        (None, "https://exa.ai"),
        ("   ", "https://exa.ai"),
    ],
)
def test_normalize_web_search_base_url_defaults_when_blank(
    base_url, default: str
):
    normalized = normalize_web_search_base_url(base_url, default=default)

    # Should return the normalized value of the provided default when base_url is blank/None
    assert normalized == normalize_web_search_base_url(default, default=default)


@pytest.mark.parametrize(
    ("base_url", "expected_message"),
```
</issue_to_address>

### Comment 2
<location path="docs/en/use/websearch.md" line_range="43" />
<code_context>
+
+Go to [Exa](https://dashboard.exa.ai) to get an API Key, then fill it in the corresponding configuration item.
+
 If you use Tavily as your web search source, you will get a better experience optimization on AstrBot ChatUI, including citation source display and more:

 ![](https://files.astrbot.app/docs/source/images/websearch/image1.png)
</code_context>
<issue_to_address>
**suggestion (typo):** “a better experience optimization” 这一表达不太自然，建议调整为更地道的英语搭配。

建议将该短语改为更自然的表达，例如使用 “a better optimized experience” 或直接 “a better experience”，如："you will get a better (optimized) experience on AstrBot ChatUI"。

```suggestion
If you use Tavily as your web search source, you will get a better experience on AstrBot ChatUI, including citation source display and more:
```
</issue_to_address>

### Comment 3
<location path="astrbot/core/tools/web_search_tools.py" line_range="384" />
<code_context>
             ]


+async def _exa_search(
+    provider_settings: dict,
+    payload: dict,
</code_context>
<issue_to_address>
**issue (complexity):** Consider introducing shared helper functions for Exa HTTP calls and POST+timeout handling to remove duplication while keeping behavior unchanged.

You can reduce the new complexity without changing behavior by introducing two small helpers:

1. **Unify Exa HTTP calls**

The three Exa helpers are almost identical except for path, action, and result mapping. A small internal helper keeps all behavior but removes repetition and makes future changes safer:

```python
async def _exa_request(
    provider_settings: dict,
    path: str,
    payload: dict,
    action: str,
    timeout: int = MIN_WEB_SEARCH_TIMEOUT,
) -> dict:
    exa_key = await _EXA_KEY_ROTATOR.get(provider_settings)
    url = f"{_get_exa_base_url(provider_settings)}{path}"
    headers = {
        "x-api-key": exa_key,
        "Content-Type": "application/json",
    }
    async with aiohttp.ClientSession(trust_env=True) as session:
        async with session.post(
            url,
            json=payload,
            headers=headers,
            timeout=_normalize_timeout(timeout),
        ) as response:
            if response.status != 200:
                reason = await response.text()
                raise Exception(
                    _format_provider_request_error(
                        "Exa", action, url, reason, response.status
                    )
                )
            return await response.json()
```

Then the three public helpers become focused on payload + mapping:

```python
async def _exa_search(
    provider_settings: dict,
    payload: dict,
    timeout: int = MIN_WEB_SEARCH_TIMEOUT,
) -> list[SearchResult]:
    data = await _exa_request(
        provider_settings,
        "/search",
        payload,
        action="web search",
        timeout=timeout,
    )
    return [
        SearchResult(
            title=item.get("title", ""),
            url=item.get("url", ""),
            snippet=(item.get("text") or "")[:500],
        )
        for item in data.get("results", [])
    ]


async def _exa_extract(
    provider_settings: dict,
    payload: dict,
    timeout: int = MIN_WEB_SEARCH_TIMEOUT,
) -> list[dict]:
    data = await _exa_request(
        provider_settings,
        "/contents",
        payload,
        action="content extraction",
        timeout=timeout,
    )
    return data.get("results", [])


async def _exa_find_similar(
    provider_settings: dict,
    payload: dict,
    timeout: int = MIN_WEB_SEARCH_TIMEOUT,
) -> list[SearchResult]:
    data = await _exa_request(
        provider_settings,
        "/findSimilar",
        payload,
        action="find similar",
        timeout=timeout,
    )
    return [
        SearchResult(
            title=item.get("title", ""),
            url=item.get("url", ""),
            snippet=(item.get("text") or "")[:500],
        )
        for item in data.get("results", [])
    ]
```

2. **Deduplicate timeout plumbing for HTTP calls**

You now repeat `timeout=_normalize_timeout(timeout)` in many places. A very small wrapper keeps the new timeout behavior but centralizes it:

```python
async def _post_json(
    session: aiohttp.ClientSession,
    url: str,
    *,
    json: dict,
    headers: dict,
    timeout: int | float | str | None = None,
):
    return await session.post(
        url,
        json=json,
        headers=headers,
        timeout=_normalize_timeout(timeout),
    )
```

Usage example (Tavily/BoCha/Baidu/Exa):

```python
async with aiohttp.ClientSession(trust_env=True) as session:
    async with _post_json(
        session,
        url,
        json=payload,
        headers=header,
        timeout=timeout,
    ) as response:
        ...
```

This keeps all existing behavior (including the minimum timeout and trust_env) but removes a lot of repeated arguments and makes future changes to timeout handling or session usage localized.
</issue_to_address>

### Comment 4
<location path="astrbot/core/utils/web_search_utils.py" line_range="49" />
<code_context>
+    return normalized
+
+
+def _iter_web_search_result_items(
+    accumulated_parts: list[dict[str, Any]],
+):
</code_context>
<issue_to_address>
**issue (complexity):** Consider inlining the small helper functions and simplifying the adapter function to reduce indirection and keep closely related logic together.

You can trim a layer or two without losing clarity or behavior.

### 1. Inline `_iter_web_search_result_items` into `collect_web_search_ref_items`

`_iter_web_search_result_items` is only used once and its logic is straightforward. Inlining it makes the flow easier to follow without losing readability:

```python
def collect_web_search_ref_items(
    accumulated_parts: list[dict[str, Any]],
    favicon_cache: dict[str, str] | None = None,
) -> list[dict[str, Any]]:
    web_search_refs: list[dict[str, Any]] = []
    seen_indices: set[str] = set()

    for part in accumulated_parts:
        if part.get("type") != "tool_call" or not part.get("tool_calls"):
            continue

        for tool_call in part["tool_calls"]:
            if (
                tool_call.get("name") not in WEB_SEARCH_REFERENCE_TOOLS
                or not tool_call.get("result")
            ):
                continue

            result = tool_call["result"]
            try:
                result_data = json.loads(result) if isinstance(result, str) else result
            except json.JSONDecodeError:
                continue

            if not isinstance(result_data, dict):
                continue

            for item in result_data.get("results", []):
                if not isinstance(item, dict):
                    continue

                ref_index = item.get("index")
                if not ref_index or ref_index in seen_indices:
                    continue

                payload = {
                    "index": ref_index,
                    "url": item.get("url"),
                    "title": item.get("title"),
                    "snippet": item.get("snippet"),
                }
                if favicon_cache and payload["url"] in favicon_cache:
                    payload["favicon"] = favicon_cache[payload["url"]]

                web_search_refs.append(payload)
                seen_indices.add(ref_index)

    return web_search_refs
```

That removes one indirection while keeping the logic in a single, obvious place.

### 2. Collapse `_extract_ref_indices` into `build_web_search_refs`

The regex + ordering behavior is easier to understand if the extraction and selection live together:

```python
def build_web_search_refs(
    accumulated_text: str,
    accumulated_parts: list[dict[str, Any]],
    favicon_cache: dict[str, str] | None = None,
) -> dict:
    ordered_refs = collect_web_search_ref_items(accumulated_parts, favicon_cache)
    if not ordered_refs:
        return {}

    refs_by_index = {ref["index"]: ref for ref in ordered_refs}

    # Inline _extract_ref_indices
    ref_indices: list[str] = []
    seen_indices: set[str] = set()
    for match in re.finditer(r"<ref>(.*?)</ref>", accumulated_text):
        ref_index = match.group(1).strip()
        if not ref_index or ref_index in seen_indices:
            continue
        ref_indices.append(ref_index)
        seen_indices.add(ref_index)

    used_refs = [refs_by_index[idx] for idx in ref_indices if idx in refs_by_index]
    if not used_refs:
        used_refs = ordered_refs

    return {"used": used_refs}
```

This keeps the “how indices are extracted” and the “how refs are ordered/fallback” logic co-located.

### 3. Simplify `collect_web_search_results` (or consider removing it)

If you still need `collect_web_search_results`, it can be a very thin adapter, making the relationship to `collect_web_search_ref_items` obvious:

```python
def collect_web_search_results(accumulated_parts: list[dict[str, Any]]) -> dict:
    return {
        ref["index"]: {
            "url": ref.get("url"),
            "title": ref.get("title"),
            "snippet": ref.get("snippet"),
        }
        for ref in collect_web_search_ref_items(accumulated_parts)
    }
```

If the only caller is effectively doing this transformation, you could alternatively have that caller perform the mapping itself and drop `collect_web_search_results` entirely to reduce the public API surface.
</issue_to_address>